package com.hjy.backend.dataManager.logger;

import com.google.common.primitives.Bytes;
import com.hjy.backend.commons.SubArray;
import com.hjy.backend.dataManager.dataItem.DataItem;
import com.hjy.backend.dataManager.page.Page;
import com.hjy.backend.dataManager.page.PageNormal;
import com.hjy.backend.dataManager.pageCache.PageCache;
import com.hjy.backend.transactionManager.TransactionManager;
import com.hjy.backend.utils.Panic;
import com.hjy.backend.utils.Parser;

import java.util.*;
import java.util.List;

/**
 * 日志文件撤销, 恢复, 重做工具类, 通过每条日志的 data 部分
 * DM 为上层模块，提供了两种操作，分别是插入新数据（I）和更新现有数据（U）
 * 对于两种数据操作，DM 记录的日志如下：
 * <p>
 * (Ti, I, A, x)，表示事务 Ti 在 A 位置插入了一条数据 x
 * (Ti, U, A, oldx, newx)，表示事务 Ti 将 A 位置的数据，从 oldx 更新成 newx
 * <p>
 * updateLog:
 * [LogType] [XID] [UID] [OldRaw] [NewRaw]
 * 对应日志类型, 事务编号, dataItem 的UID, 旧数据, 新数据
 * <p>
 * insertLog:
 * [LogType] [XID] [pageId] [Offset] [Raw]
 * 对应日志类型, 事务编号, 数据页编号, 偏移, 数据
 * <p>
 * 我们有两个规定:
 * 规定1：正在进行的事务，不会读取其他任何未提交的事务产生的数据。
 * 规定2：正在进行的事务，不会修改其他任何未提交的事务修改或产生的数据。
 * 在 JyDB 中，由于 VersionManager 的存在，传递到 DataManager 层，真正执行的操作序列，都可以保证规定 1 和规定 2
 * <p>
 * 日志的恢复:
 * 1. 重做所有崩溃时已完成（committed 或 aborted）的事务
 * 2. 撤销所有崩溃时未完成（active）的事务
 */
public class Recover {
    // 日志类型
    private static final byte LOG_TYPE_INSERT = 0;
    private static final byte LOG_TYPE_UPDATE = 1;

    // 重做还是撤销
    private static final int REDO = 0;
    private static final int UNDO = 1;

    /**
     * 新增型日志类
     * [LogType] [XID] [PageID] [Offset] [Raw]
     */
    static class InsertLogInfo {
        long xid;
        int pageID;
        short offset;
        byte[] raw;
    }

    /**
     * 修改型日志类
     * [LogType] [XID] [UID] [OldRaw] [NewRaw]
     */
    static class UpdateLogInfo {
        long xid;
        int pageID;
        short offset;
        byte[] oldRaw;
        byte[] newRaw;
    }

    /**
     * 恢复日志, 入口方法
     */
    public static void recover(TransactionManager tm, Logger lg, PageCache pc) {
        System.out.println("Recover(日志恢复中)...");
        // 开始读取日志
        lg.rewind();
        int maxPageId = 0;

        // 更新最大页面个数
        while (true) {
            // 获取数据部分
            byte[] log = lg.next();
            if (log == null) break;

            // 获取日志的 pageId
            int pageId;
            if (isInsertLog(log)) {
                InsertLogInfo insertLog = parseInsertLog(log);
                pageId = insertLog.pageID;
            } else {
                UpdateLogInfo updateLog = parseUpdateLog(log);
                pageId = updateLog.pageID;
            }

            if (pageId > maxPageId) {
                maxPageId = pageId;
            }
        }

        if (maxPageId == 0) {
            maxPageId = 1;
        }

        // 根据最大页面个数, 截断数据文件
        pc.truncateByPageId(maxPageId);
        System.out.println("Truncate to " + maxPageId + " pages.");

        // 重做事务
        redoTransactions(tm, lg, pc);
        System.out.println("Redo Transactions Over(重做事务完成).");

        // 撤销事务
        undoTransactions(tm, lg, pc);
        System.out.println("Undo Transactions Over(撤销事务完成).");

        System.out.println("Recovery Over(日志文件恢复完毕)");
    }

    /**
     * 重做事务,正序扫描事务 T 的所有日志
     */
    private static void redoTransactions(TransactionManager tm, Logger lg, PageCache pc) {
        lg.rewind();
        while (true) {
            byte[] log = lg.next();
            if (log == null) break;
            if (isInsertLog(log)) {
                InsertLogInfo insertLog = parseInsertLog(log);
                long xid = insertLog.xid;
                // 事务都提交了, 就不要重做了
                if (!tm.isActive(xid)) {
                    doInsertLog(pc, log, REDO);
                }
            } else {
                UpdateLogInfo updateLog = parseUpdateLog(log);
                long xid = updateLog.xid;
                // 事务都提交了, 就不要重做了
                if (!tm.isActive(xid)) {
                    doUpdateLog(pc, log, REDO);
                }
            }
        }
    }

    /**
     * 撤销事务, 倒序扫描事务 T 的所有日志
     */
    private static void undoTransactions(TransactionManager tm, Logger lg, PageCache pc) {
        Map<Long, List<byte[]>> logCache = new HashMap<>();
        lg.rewind();

        // 将每个事务对应的日志整合成一个 list
        while (true) {
            byte[] log = lg.next();
            if (log == null) break;
            if (isInsertLog(log)) {
                InsertLogInfo insertLog = parseInsertLog(log);
                long xid = insertLog.xid;
                // 如果这个事务还没有提交, 说明上一次数据库启动有点问题, 那就撤销这次事务的提交
                if (tm.isActive(xid)) {
                    if (!logCache.containsKey(xid)) {
                        logCache.put(xid, new ArrayList<>());
                    }
                    logCache.get(xid).add(log);
                }
            } else {
                UpdateLogInfo updateLog = parseUpdateLog(log);
                long xid = updateLog.xid;
                if (tm.isActive(xid)) {
                    if (!logCache.containsKey(xid)) {
                        logCache.put(xid, new ArrayList<>());
                    }
                    logCache.get(xid).add(log);
                }
            }
        }


        // 对所有的active log进行倒序undo, 因为倒序可以保证数据一致性
        for (Map.Entry<Long, List<byte[]>> entry : logCache.entrySet()) {
            List<byte[]> logs = entry.getValue();
            for (int i = logs.size() - 1; i >= 0; i--) {
                byte[] log = logs.get(i);
                if (isInsertLog(log)) {
                    doInsertLog(pc, log, UNDO);
                } else {
                    doUpdateLog(pc, log, UNDO);
                }
            }
            tm.abort(entry.getKey());
        }
    }

    private static boolean isInsertLog(byte[] log) {
        return log[0] == LOG_TYPE_INSERT;
    }

    // 修改型日志相关的方法如下 [LogType] [XID] [UID] [OldRaw] [NewRaw]
    private static final int OF_TYPE = 0; // 0
    private static final int OF_XID = OF_TYPE + 1; // 1
    private static final int OF_UPDATE_UID = OF_XID + 8; // 9
    private static final int OF_UPDATE_RAW = OF_UPDATE_UID + 8; // 17

    /**
     * 创建一个修改型的日志
     *
     * @param xid
     * @param di
     * @return
     */
    public static byte[] updateLog(long xid, DataItem di) {
        byte[] logType = {LOG_TYPE_UPDATE}; // 日志类型
        byte[] xidRaw = Parser.long2Byte(xid); // 事务编号
        byte[] uidRaw = Parser.long2Byte(di.getUid()); // dataItem 的 UID
        byte[] oldRaw = di.getOldRaw(); // 旧数据
        SubArray raw = di.getRaw();
        byte[] newRaw = Arrays.copyOfRange(raw.raw, raw.start, raw.end); // 新数据
        return Bytes.concat(logType, xidRaw, uidRaw, oldRaw, newRaw);
    }

    /**
     * 解析一个日志为修改日志格式
     *
     * @param log
     * @return
     */
    private static UpdateLogInfo parseUpdateLog(byte[] log) {
        UpdateLogInfo updateLog = new UpdateLogInfo();
        // 第一位是日志类型
        // 事务号
        updateLog.xid = Parser.parseLong(Arrays.copyOfRange(log, OF_XID, OF_UPDATE_UID)); // 1~9
        // dataItem 的 UID, UID 存储着 PageId 和 offset
        long uid = Parser.parseLong(Arrays.copyOfRange(log, OF_UPDATE_UID, OF_UPDATE_RAW)); // 9~17
        // 逻辑与 uid 与 1111 1111 1111 1111 1111 1111 1111 1111 进行与
        updateLog.offset = (short) (uid & ((1L << 16) - 1));
        // 无符号右移
        uid >>>= 32;
        // 逻辑与 uid 与 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 进行与
        updateLog.pageID = (int) (uid & ((1L << 32) - 1));
        // 一个 raw 的长度
        int length = (log.length - OF_UPDATE_RAW) / 2;
        updateLog.oldRaw = Arrays.copyOfRange(log, OF_UPDATE_RAW, OF_UPDATE_RAW + length); // 17~17+len
        updateLog.newRaw = Arrays.copyOfRange(log, OF_UPDATE_RAW + length, OF_UPDATE_RAW + length * 2); // 17+len~17+len*2
        return updateLog;
    }

    /**
     * 重做或撤销修改型的日志
     * <p>
     * 重做: 就将 A 位置的值设置为 newx
     * <p>
     * 撤销: 将 对应 位置的值设置为 oldx
     *
     * @param pc
     * @param log
     * @param flag
     */
    private static void doUpdateLog(PageCache pc, byte[] log, int flag) {
        int pgno;
        short offset;
        byte[] raw;
        if (flag == REDO) {
            UpdateLogInfo xi = parseUpdateLog(log);
            pgno = xi.pageID;
            offset = xi.offset;
            raw = xi.newRaw;
        } else {
            UpdateLogInfo xi = parseUpdateLog(log);
            pgno = xi.pageID;
            offset = xi.offset;
            raw = xi.oldRaw;
        }
        Page pg = null;
        try {
            pg = pc.getPage(pgno);
        } catch (Exception e) {
            Panic.panic(e);
        }
        try {
            // 写入数据页
            PageNormal.recoverUpdate(pg, raw, offset);
        } finally {
            pg.release();
        }
    }

    // 新增型日志相关的方法如下 [LogType] [XID] [PageID] [Offset] [Raw]
    private static final int OF_INSERT_PAGE_ID = OF_XID + 8;
    private static final int OF_INSERT_OFFSET = OF_INSERT_PAGE_ID + 4;
    private static final int OF_INSERT_RAW = OF_INSERT_OFFSET + 2;

    /**
     * 创建一个新增型的日志
     *
     * @param xid
     * @param pg
     * @param raw
     * @return
     */
    public static byte[] insertLog(long xid, Page pg, byte[] raw) {
        byte[] logTypeRaw = {LOG_TYPE_INSERT}; // 日志类型
        byte[] xidRaw = Parser.long2Byte(xid); // 事务编号
        byte[] PageIdRaw = Parser.int2Byte(pg.getPageNumber()); // 数据页编号
        byte[] offsetRaw = Parser.short2Byte(PageNormal.getFSO(pg)); // 偏移地址
        return Bytes.concat(logTypeRaw, xidRaw, PageIdRaw, offsetRaw, raw);
    }

    /**
     * 解析新增型日志
     *
     * @param log
     * @return
     */
    private static InsertLogInfo parseInsertLog(byte[] log) {
        InsertLogInfo li = new InsertLogInfo();
        li.xid = Parser.parseLong(Arrays.copyOfRange(log, OF_XID, OF_INSERT_PAGE_ID));
        li.pageID = Parser.parseInt(Arrays.copyOfRange(log, OF_INSERT_PAGE_ID, OF_INSERT_OFFSET));
        li.offset = Parser.parseShort(Arrays.copyOfRange(log, OF_INSERT_OFFSET, OF_INSERT_RAW));
        li.raw = Arrays.copyOfRange(log, OF_INSERT_RAW, log.length);
        return li;
    }

    /**
     * 重做或撤销新增型的日志
     * <p>
     * 重做: 重新插入 对应 位置
     * <p>
     * 撤销: 将 对应 DataItem的ValidFlag设置为 不可使用
     *
     * @param pc
     * @param log
     * @param flag
     */
    private static void doInsertLog(PageCache pc, byte[] log, int flag) {
        InsertLogInfo li = parseInsertLog(log);
        Page pg = null;
        try {
            pg = pc.getPage(li.pageID);
        } catch (Exception e) {
            Panic.panic(e);
        }
        try {
            if (flag == UNDO) {
                DataItem.setDataItemRawInvalid(li.raw);
            }
            // li.raw 是一个DataItem 数据
            PageNormal.recoverInsert(pg, li.raw, li.offset);
        } finally {
            pg.release();
        }
    }
}
